%% copy riak-0.14.2 source to learn dynamo implementation and erlang
%% author: lichuang
%% copy source from riak_core_ring_handler.erl
-module(edynamo_ring_handler).

-behaviour(gen_event).

%% gen_event callbacks
-export([init/1, handle_event/2, handle_call/2,
         handle_info/2, terminate/2, code_change/3]).
-record(state, {}).

%% ===================================================================
%% gen_event callbacks
%% ===================================================================
init([]) ->
  {ok, Ring} = edynamo_ring_manager:get_my_ring(),
  ensure_vnodes_started(Ring),
  {ok, #state{}}.

handle_event({ring_update, Ring}, State) ->
  ensure_vnodes_started(Ring),
  {ok, State}.

handle_call(_Event, State) ->
    {ok, ok, State}.

handle_info(_Info, State) ->
    {ok, State}.

terminate(_Reason, _State) ->
    ok.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% ===================================================================
%% Internal functions
%% ===================================================================
ensure_vnodes_started(Ring) ->
  case edynamo_core:vnode_modules() of
    [] -> ok;
    Mods ->
      case ensure_vnodes_started(Mods, Ring, []) of
	[] -> edynamo_core:stop("node removal completed, exiting.");
	_  -> ok
      end
  end.

ensure_vnodes_started([], _Ring, Acc) ->
  lists:flatten(Acc);
ensure_vnodes_started([H|T], Ring, Acc) ->
  ensure_vnodes_started(T, Ring, [ensure_vnodes_started(H, Ring) | Acc]).

ensure_vnodes_started(Mod, Ring) ->
  Startable = startable_vnodes(Mod, Ring),
  [Mod:start_vnode(I) || I <- Startable],
  Startable.

startable_vnodes(Mod, Ring) ->
  AllMembers = edynamo_ring:all_members(Ring),
  case {length(AllMembers), hd(AllMembers) =:= node()} of
    {1, true} ->
      edynamo_ring:my_indices(Ring);
    _ ->
      {ok, Excl} = edynamo_handoff_manager:get_exclusions(Mod),
      case edynamo_ring:random_other_index(Ring, Excl) of
	no_indices ->
	  case length(Excl) =:= edynamo_ring:num_partitions(Ring) of
	    true ->
	      [];
	    false ->
	      edynamo_ring:my_indices(Ring)
	  end;
	RO ->
	  [RO | edynamo_ring:my_indices(Ring)]
      end
  end.
